package com.cellulam.framework.datasource.service;


import com.cellulam.framework.core.model.*;
import com.cellulam.framework.core.utils.BeanUtil;
import com.cellulam.framework.core.utils.ClassUtil;
import com.cellulam.framework.core.utils.JacksonUtils;
import com.cellulam.framework.datasource.FetchDataBean;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.io.Serializable;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * 基础DTOService
 */
@Slf4j
public abstract class BaseDtoService<PK, BEAN extends Serializable, REQ extends Serializable, RESP extends Serializable>
        extends BaseCrudService<PK, BEAN> {

    protected Class<REQ> reqClass;
    protected Class<RESP> respClass;

    public BaseDtoService() {
        reqClass = ClassUtil.getSuperClassGenericType(getClass(), 2);
        respClass = ClassUtil.getSuperClassGenericType(getClass(), 3);
    }

    /**
     * 保存<code>request</code>
     *
     * @param request
     * @return 新增的数据的行主键(如果主键是自增)
     */
    public abstract PK save(REQ request);

    /**
     * 通过主键获取一行数据, 并转换为DTOResponse
     *
     * @param id
     * @return DTOResponse
     */
    public RESP getById(PK id) {
        BEAN bean = getBeanById(id);
        return BeanUtil.copy(bean, respClass);
    }

//    /**
//     * 保存<code>request</code>
//     * @param request
//     * @return 新增的数据的行主键(如果主键是自增)
//     */
//    public PK save(REQ request) {
//        BEAN bean = BeanUtil.copy(request, entityClass);
//        saveBean(bean);
//        request.setApplicationID(bean.getApplicationID());
//        return bean.getApplicationID();
//    }

    /**
     * 通过主键来更新数据表行
     *
     * @param request
     * @return 影响数据表的行数
     */
    public int update(REQ request) {
        BEAN bean = BeanUtil.copy(request, entityClass);
        return updateBean(bean);
    }

    public PagerResponse<RESP> query(PagerRequest<REQ> pagerRequest) {
        PagerResponse<RESP> result = new PagerResponse<>();
        result.setPageSize(pagerRequest.getPageSize());
        BEAN criteria = BeanUtil.copy(pagerRequest.getCriteria(), entityClass);
        PagerCriteria<BEAN> pagerCriteria = pagerRequest.toCriteria(criteria);
        long count = getCrudDao().count(pagerCriteria);
        if (count >= 0) {
            result.setTotal(count);
            result.setData(BeanUtil.copyByList(getCrudDao().query(pagerCriteria), respClass));
        }
        return result;
    }

    public LimitResponse<RESP> queryPage(PagerRequest<REQ> pagerRequest) {
        return this.queryPage(pagerRequest, respClass);
    }

    public <R> LimitResponse<R> queryPage(PagerRequest<REQ> pagerRequest, Class<R> responseClass) {
        BEAN criteria = BeanUtil.copy(pagerRequest.getCriteria(), entityClass);
        PagerCriteria<BEAN> pagerCriteria = pagerRequest.toCriteria(criteria);

        int pageSize = pagerCriteria.getLimit();
        pagerCriteria.setLimit(pageSize + 1);

        List<R> resp = BeanUtil.copyByList(this.getCrudDao().query(pagerCriteria), responseClass);
        return new LimitResponse<>(pageSize, resp);
    }

    public LimitResponse<RESP> query(int page, int pageSize, REQ req) {
        int offset = (page - 1) * pageSize;

        BEAN criteria = BeanUtil.copy(req, entityClass);
        List<RESP> resp = BeanUtil.copyByList(this.findBeans(criteria, offset, pageSize + 1), respClass);

        return new LimitResponse<>(pageSize, resp);
    }

    /**
     * 执行 查询
     *
     * @param criteria
     * @return list of DTOResponse(s)
     */
    public RESP findOnResult(REQ criteria) {
        BEAN cri = BeanUtil.copy(criteria, entityClass);
        return BeanUtil.copy(super.findOneBean(cri), respClass);
    }

    /**
     * 执行 查询
     *
     * @param criteria
     * @return list of DTOResponse(s)
     */
    public List<RESP> find(REQ criteria) {
        BEAN cri = BeanUtil.copy(criteria, entityClass);
        return BeanUtil.copyByList(findBeans(cri), respClass);
    }

    /**
     * 批量处理
     * 不抛出异常
     *
     * @param criteria
     * @param consumer
     */
    public void batchProcessSilence(BEAN criteria, Consumer<BEAN> consumer) {
        batchProcess(criteria, consumer, null);
    }

    /**
     * 批量处理
     *
     * @param criteria
     * @param consumer
     */
    public void batchProcess(BEAN criteria, Consumer<BEAN> consumer, Consumer<Exception> exceptionHandler) {
        batchProcess(criteria, x -> {
            try {
                consumer.accept(x);
            } catch (Exception e) {
                if (exceptionHandler == null) {
                    log.error("batch process data error, data: {}, e",
                            JacksonUtils.toJson(x), e);
                } else {
                    exceptionHandler.accept(e);
                }
            }
        });
    }

    /**
     * 批量处理
     *
     * @param criteria
     * @param consumer
     */
    public void batchProcess(BEAN criteria, Consumer<BEAN> consumer) {
        long start = 0;
        int limit = 500;

        while (true) {
            List<FetchDataBean<BEAN>> data = this.fetch(start, limit, criteria);
            if (CollectionUtils.isEmpty(data)) {
                break;
            }

            if (this.batchParallel()) {
                data.parallelStream()
                        .forEach(x -> consumer.accept(x.getBean()));
            } else {
                data.stream()
                        .forEach(x -> consumer.accept(x.getBean()));
            }

            start = data.stream().mapToLong(x -> x.getPk()).max().orElse(0);
        }
    }


    /**
     * 批量处理，默认只返回错误数据
     *
     * @param criteria
     * @param function
     * @param <T>
     * @return
     */
    public <T> BatchProcessResult<T> batchSubmit(BEAN criteria, Function<BEAN, ResultData<T>> function) {
        return this.batchSubmit(criteria, function, true);
    }

    /**
     * 批量处理
     *
     * @param criteria
     * @param function
     * @param onlyFail 是否只返回错误数据
     * @param <T>
     * @return
     */
    public <T> BatchProcessResult<T> batchSubmit(BEAN criteria, Function<BEAN, ResultData<T>> function, boolean onlyFail) {
        List<T> success = Lists.newCopyOnWriteArrayList();
        List<T> fail = Lists.newCopyOnWriteArrayList();

        this.batchProcess(criteria, x -> {
            ResultData<T> result = function.apply(x);
            if (result.isSuccess() && !onlyFail) {
                success.add(result.getData());
            } else if (!result.isSuccess()) {
                fail.add(result.getData());
            }
        });

        return BatchProcessResult.<T>builder()
                .success(success)
                .fail(fail)
                .build();
    }

    /**
     * 是否并行处理
     *
     * @return
     */
    protected boolean batchParallel() {
        return true;
    }
}
